package com.example.mq.kafka.producer;

import com.alibaba.fastjson.JSONObject;
import com.example.mq.kafka.dataobject.Kafka01Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;

/**
 * @author luowj12
 * @since 2021/4/25 10:04
 */
@Component
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<Object, Object> kafkaTemplate;

    public SendResult syncSend(Integer id) throws ExecutionException, InterruptedException {
        // 创建 Demo01Message 消息
        Kafka01Message message = new Kafka01Message();
        message.setId(id);
        // 同步发送消息
        return kafkaTemplate.send(Kafka01Message.TOPIC, message).get();
    }

    public ListenableFuture<SendResult<Object, Object>> asyncSend(Integer id) {
        // 创建 Demo01Message 消息
        Kafka01Message message = new Kafka01Message();
        message.setId(id);
        // 异步发送消息
        return kafkaTemplate.send(Kafka01Message.TOPIC, message);
    }

    /**
     * 默认分词器 根据键的哈希值来确定消息发送到哪个分区
     * @param id
     * @return
     */
    public ListenableFuture<SendResult<Object, Object>> asyncSendPartition(Integer id) {
        // 创建 Demo01Message 消息
        Kafka01Message message = new Kafka01Message();
        message.setId(id);
        // 异步发送消息
        return kafkaTemplate.send(Kafka01Message.TOPIC,String.valueOf(id), message);
    }

}
